package com.afdor.rws.event.driven.support;

import com.afdor.rws.annotation.extension.Extension;
import com.afdor.rws.core.Callback;
import com.afdor.rws.core.extension.ExtensionLoader;
import com.afdor.rws.core.strategy.IdGeneratorStrategy;
import com.afdor.rws.event.driven.EventBus;
import com.afdor.rws.event.driven.EventChangeListener;
import com.afdor.rws.event.driven.EventChangeManager;
import com.afdor.rws.event.driven.EventPersistenceManager;
import com.afdor.rws.event.driven.EventSource;
import com.afdor.rws.event.driven.EventSourceManager;
import com.afdor.rws.event.driven.channel.ChannelProvider;
import com.afdor.rws.event.driven.channel.consumer.handler.EventHandler;
import com.afdor.rws.event.driven.channel.consumer.handler.EventRetryStrategy;
import com.afdor.rws.event.driven.commons.ThreadUtils;
import com.afdor.rws.event.driven.config.EventBusConfig;
import com.afdor.rws.event.driven.config.EventConsumerConfig;
import com.afdor.rws.event.driven.entities.Event;
import com.afdor.rws.event.driven.entities.holder.CallbackEvent;
import com.afdor.rws.event.driven.entities.holder.ResultEvent;
import com.afdor.rws.event.driven.support.transaction.EventBusTransactionItem;
import com.afdor.rws.event.driven.support.transaction.impl.DefaultEventBusTransactionItemImpl;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;

/**
 * 事件中线默认实现
 *
 * @author 悭梵
 * @date Created in 2018-08-14 15:08
 */
@Slf4j
@Extension(value = "default", order = Extension.NORMAL_ORDER - 100, interfaceClass = EventBus.class)
public class DefaultEventBus implements EventBus {

    private final AtomicBoolean init = new AtomicBoolean();

    /**
     * 配置信息
     */
    private EventBusConfig config;
    private EventSourceManager eventSourceManager;
    private EventPersistenceManager eventPersistenceManager;
    private EventChangeManager eventStatusChangeManager;

    /**
     * 事物提交队列管理器
     */
    private ConcurrentMap<String, EventBusTransactionItem> eventTransactionMap = new ConcurrentHashMap<String, EventBusTransactionItem>(128);

    @Override
    public <T extends EventBus> T init() {
        /* jvm 退出执行方法 */
        Runtime.getRuntime().addShutdownHook(new Thread(() -> DefaultEventBus.this.destroy()));
        if (Objects.nonNull(getEventSourceManager())) {
            getEventSourceManager().init();
        }
        if (Objects.nonNull(getEventStatusChangeManager())) {
            getEventStatusChangeManager().init();
        }
        if (Objects.nonNull(getEventPersistenceManager())) {
            getEventPersistenceManager().init();
        }
        init.compareAndSet(Boolean.FALSE, Boolean.TRUE);
        return (T) this;
    }

    @Override
    public void destroy() {
        init.compareAndSet(Boolean.TRUE, Boolean.FALSE);
        if (Objects.nonNull(getEventStatusChangeManager())) {
            getEventStatusChangeManager().destroy();
        }
        if (Objects.nonNull(getEventPersistenceManager())) {
            getEventPersistenceManager().destroy();
        }
        if (Objects.nonNull(getEventSourceManager())) {
            getEventSourceManager().destroy();
        }
    }

    @Override
    public <C extends EventBusConfig> C getConfig() {
        return (C) this.config;
    }

    @Override
    public <T extends EventBus, C extends EventBusConfig> T setConfig(C config) {
        this.config = config;
        return (T) this;
    }

    @Override
    public EventSourceManager getEventSourceManager() {
        return this.eventSourceManager;
    }

    @Override
    public <T extends EventBus> T setEventSourceManager(@NonNull EventSourceManager eventSourceManager) {
        this.eventSourceManager = eventSourceManager;
        return (T) this;
    }

    @Override
    public EventPersistenceManager getEventPersistenceManager() {
        return this.eventPersistenceManager;
    }

    @Override
    public <T extends EventBus> T setEventPersistenceManager(@NonNull EventPersistenceManager eventPersistenceManager) {
        this.eventPersistenceManager = eventPersistenceManager;
        return (T) this;
    }

    @Override
    public EventChangeManager getEventStatusChangeManager() {
        return this.eventStatusChangeManager;
    }

    @Override
    public <T extends EventBus> T setEventStatusChangeManager(@NonNull EventChangeManager eventStatusChangeManager) {
        this.eventStatusChangeManager = eventStatusChangeManager;
        return (T) this;
    }

    @Override
    public <T extends EventBus> T addEventStatusChangeListener(@NonNull EventChangeListener listener) {
        EventChangeManager eventStatusChangeManager = getEventStatusChangeManager();
        if (eventStatusChangeManager != null) {
            eventStatusChangeManager.addEventChangeListener(listener);
        }
        return (T) this;
    }

    @Override
    public <T extends EventBus> T registerHandler(@NonNull String eventSourceName, @NonNull EventConsumerConfig eventConsumerConfig, @NonNull EventHandler eventHandler) {
        return registerHandler(eventSourceName, eventConsumerConfig, eventHandler, null);
    }

    @Override
    public <T extends EventBus> T registerHandler(@NonNull String eventSourceName, @NonNull EventConsumerConfig eventConsumerConfig, @NonNull EventHandler eventHandler, EventRetryStrategy eventRetryStrategy) {
        checkup();

        if (StringUtils.isBlank(eventSourceName)) {
            throw new NullPointerException("未注册处理器到对应事件源的名称不能为空");
        }

        EventSource eventSource = (EventSource) Optional.ofNullable(getEventSourceManager().getEventSource(eventSourceName))
                .orElseThrow(() -> new NullPointerException("未获取到对应的事件源" + eventSourceName + "配置项"));

        Optional.ofNullable(eventSource.getChannelProvider())
                .orElseThrow(() -> new NullPointerException("事件处理器对应的发送通道为空"))
                .subscribe(eventConsumerConfig, eventHandler, eventRetryStrategy);

        return (T) this;
    }

    @Override
    public <V> ResultEvent<V> sendEvent(@NonNull Event event) {

        EventSource eventSource = (EventSource) Optional.ofNullable(getEventSourceManager().getEventSource(event.getEventSourceName()))
                .orElseThrow(() -> new NullPointerException("未获取到对应的事件源" + event.getEventSourceName() + "配置项"));

        ChannelProvider channelProvider = Optional.ofNullable(eventSource.getChannelProvider())
                .orElseThrow(() -> new NullPointerException("事件处理器对应的发送通道为空"));

        event.setSendType(Event.SendType.SYNC);
        event.setStatus(Event.Status.EXECUTING_EVENT_BUS);
        ResultEvent<V> result = channelProvider.sendEvent(event);

        if (result != null) {
            event.setStatus(result.getCode() == 1 ? Event.Status.COMMITTED_EVENT_BUS : Event.Status.TERMINATED_EVENT_BUS);
        }
        return result;
    }

    @Override
    public void sendAsyncEvent(@NonNull Event event) {

        EventSource eventSource = (EventSource) Optional.ofNullable(getEventSourceManager().getEventSource(event.getEventSourceName()))
                .orElseThrow(() -> new NullPointerException("未获取到对应的事件源" + event.getEventSourceName() + "配置项"));

        ChannelProvider channelProvider = Optional.ofNullable(eventSource.getChannelProvider())
                .orElseThrow(() -> new NullPointerException("事件处理器对应的发送通道为空"));

        event.setSendType(Event.SendType.ASYNC);
        event.setStatus(Event.Status.EXECUTING_EVENT_BUS);
        channelProvider.sendAsyncEvent(event);
    }

    @Override
    public <V> void asyncSendEvent(@NonNull Event event, Callback<ResultEvent<V>, Void> callback) {
        EventSource eventSource = (EventSource) Optional.ofNullable(getEventSourceManager().getEventSource(event.getEventSourceName()))
                .orElseThrow(() -> new NullPointerException("未获取到对应的事件源" + event.getEventSourceName() + "配置项"));

        ChannelProvider channelProvider = Optional.ofNullable(eventSource.getChannelProvider())
                .orElseThrow(() -> new NullPointerException("事件处理器对应的发送通道为空"));

        event.setSendType(Event.SendType.ASYNC);
        event.setStatus(Event.Status.EXECUTING_EVENT_BUS);
        channelProvider.asyncSendEvent(new CallbackEvent(event, callback));
    }

    @Override
    public void batchSendAsyncEvent(Event... events) {
        if (events == null || events.length == 0) {
            throw new NullPointerException("发送事件不能为空");
        }
        batchSendAsyncEvent(Arrays.asList(events));
    }

    @Override
    public void batchSendAsyncEvent(@NonNull List<Event> eventList) {
        eventList.stream().collect(Collectors.groupingBy(event -> {
            event.setSendType(Event.SendType.ASYNC_CALLBACK);
            event.setStatus(Event.Status.EXECUTING_EVENT_BUS);
            return event.getEventSourceName();
        }, HashMap::new, Collectors.toList()))
                .entrySet()
                .parallelStream()
                .forEach((entry) -> {
                    EventSource eventSource = (EventSource) Optional.ofNullable(getEventSourceManager().getEventSource(entry.getKey()))
                            .orElseThrow(() -> new NullPointerException("未获取到对应的事件源" + entry.getKey() + "配置项"));

                    ChannelProvider channelProvider = Optional.ofNullable(eventSource.getChannelProvider())
                            .orElseThrow(() -> new NullPointerException("事件处理器对应的发送通道为空"));

                    channelProvider.batchSendAsyncEvent(entry.getValue());
                });
    }

    @Override
    public String newTransaction() {
        checkup();

        ExtensionLoader<IdGeneratorStrategy> idGeneratorStrategyExtensionLoader = ExtensionLoader.getInstance(IdGeneratorStrategy.class);
        // 获取配置的Id生成策略
        String transactionId = String.valueOf(idGeneratorStrategyExtensionLoader.getExtensionOrDefault(getConfig().getTransactionIdStrategyName()).generate());
        if (eventTransactionMap.containsKey(transactionId)) {
            return newTransaction();
        } else {
            eventTransactionMap.putIfAbsent(transactionId, new DefaultEventBusTransactionItemImpl(transactionId, getEventSourceManager()));
        }
        return transactionId;
    }

    @Override
    public void addEvent(@NonNull final String transactionId, @NonNull final Event event) {
        EventBusTransactionItem transactionItem = getEventBusTransactionItem(transactionId);
        transactionItem.addEvent(event);
    }

    @Override
    public boolean commit(@NonNull final String transactionId) {
        final EventBusTransactionItem transactionItem = getEventBusTransactionItem(transactionId);
        return transactionItem.commit();
    }

    @Override
    public void rollback(@NonNull final String transactionId) {
        final EventBusTransactionItem transactionItem = getEventBusTransactionItem(transactionId);
        transactionItem.rollback();
    }

    /**
     * 验证
     */
    private void checkup() {
        if (Objects.isNull(getConfig())) {
            new NullPointerException("事件总线相关配置信息不能为空");
        }
        if (Objects.isNull(getEventSourceManager())) {
            new NullPointerException("事件源管理器不能为空");
        }
        if (Objects.isNull(getEventPersistenceManager())) {
            new NullPointerException("事件持久化管理器不能为空");
        }
    }

    /**
     * 获取事物项
     *
     * @param transactionId
     * @return
     */
    private EventBusTransactionItem getEventBusTransactionItem(@NonNull String transactionId) {
        EventBusTransactionItem transactionItem = eventTransactionMap.get(transactionId);
        if (transactionItem == null) {
            throw new IllegalStateException("the specified transaction is not turned on or has been executed. please open a new transaction first . ");
        }
        return transactionItem;
    }

    /**
     * 回调相关
     *
     * @param event
     * @param <V>
     * @return
     */
    private <V> CompletableFuture<ResultEvent<V>> getCompletableFuture(@NonNull final Event event) {
        CompletableFuture<ResultEvent<V>> completableFuture = new CompletableFuture<>();
        completableFuture.exceptionally((throwable) -> {
            if (throwable instanceof CancellationException) {
                event.setStatus(Event.Status.ROLLEDBACK);
            }
            ResultEvent<V> result = new ResultEvent<>(event);
            result.setCode(0);
            result.setMessage(ThreadUtils.getStackMessage(throwable));
            return result;
        });
        return completableFuture;
    }

}
